feat: add negative ack redelivery delay and backoff support (PIP-106)#419
Open
abbudao wants to merge 67 commits into
Open
feat: add negative ack redelivery delay and backoff support (PIP-106)#419abbudao wants to merge 67 commits into
abbudao wants to merge 67 commits into
Conversation
- Adds pure negative_ack_backoff module API skeleton - Covers default, custom, zero-delay, validation, and custom trait vectors - Wires consumer-module exports for the new policy types Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Validates multiplier and delay inputs with exact Error::Custom messages - Computes multiplier delays with max-delay capping and u32::MAX safety - Preserves zero-min-delay immediate redelivery behavior Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Re-exports NegativeAckBackoff from the root API - Re-exports MultiplierRedeliveryBackoff and its builder - Keeps Phase 2 API available from both consumer and crate root Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Covers 60s default fixed delay selection - Covers explicit zero-delay immediate redelivery - Covers message-count backoff and ID-only fixed fallback Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Defaults missing nack timing to a 60 second fixed delay - Applies message-count backoff only when a redelivery count is available - Preserves explicit zero-delay immediate redelivery behavior Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Covers batch index normalization with partial-batch cancellation - Covers due collection, in-flight lifecycle, retry, and clear behavior - Covers partition-scoped cumulative ack cancellation and redelivery ID field normalization Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Tracks pending nacks by normalized ledger/entry/partition keys with batch-index scopes - Cancels normal and cumulative acks without dropping sibling batch nacks - Moves due entries in-flight before dispatch and supports retry, dispatched, and clear transitions Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Locks public EngineMessage::Nack to ID-only shape - Expects a private negative-ack due event - Asserts TopicConsumer passes nack timing config to ConsumerEngine and reader/multi paths keep using TopicConsumer::new Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Keeps public EngineMessage::Nack ID-only while adding a private negative-ack due event - Carries fixed delay and backoff config from TopicConsumer into ConsumerEngine - Adds lazy tracker/ticker state fields without starting ticker work at construction time Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Expects tracker-based non-zero scheduling and zero-delay immediate redelivery - Expects lazy coalesced due ticker and retry-safe dispatch lifecycle - Expects ack cancellation and close cleanup of negative-ack tracker state Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Schedules non-zero nacks through the private tracker and keeps zero-delay nacks immediate - Starts one lazy coalescing ticker that sends private due events at the 500ms tracker interval - Drains due entries in-flight, removes them on successful sender dispatch, retries failures, and cancels on ack/cumulative ack - Preserves reconnect state while clearing tracker/ticker state on close/drop Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Assert builder rustdoc covers Phase 4 negative-ack runtime semantics and caveats - Assert public EngineMessage::Nack still accepts a default MessageIdData with one field Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Document the 60s default, zero-delay escape hatch, message-count backoff, ID-only fixed fallback, batch normalization, timer caveats, and DLQ/retry-letter boundary - Keep public EngineMessage::Nack compatibility covered with a one-field MessageIdData::default construction test Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Compile tracker inspection helpers only for tests so doc-test and clippy validation do not warn on production builds Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Avoid broad grep false positives in runtime docs while preserving the retry-letter/DLQ boundary - Address clippy warnings in negative ack tracker scope merging and ticker fallback dispatch Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Add a focused source assertion that reconnect does not clear negative ack tracker state Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Route TopicConsumer::close through the engine so pending negative-ack state and ticker flags are cleared before shutdown - Normalize immediate negative-ack redelivery IDs through the tracker helper, matching delayed due dispatch semantics - Use checked Instant arithmetic for large custom backoff delays and cover it with deterministic tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Ensure a successful zero-delay negative ack removes any existing delayed tracker entry for the same normalized broker entry, preventing duplicate later redelivery. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Route all negative ack delay selection through tracker scheduling so custom backoff implementations are invoked once per nack. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Remove nacked entries from the existing unacked-message timeout tracker using normalized redelivery-entry matching so timeout redelivery cannot race the negative-ack delay/backoff path. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Preserve broker redelivery-count absence through consumer delivery so negative-ack backoff only applies when the broker supplied a count. Stop the unacked-redelivery ticker on close/drop, keep failed immediate redeliveries retryable, and avoid over-canceling batch siblings when ack_set is present with a batch index. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Restore public consumer data aliases while keeping optional broker redelivery counts internal, tighten unacked ticker shutdown, and add an immediate nack fallback when retry ticker startup fails. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Adds not-before-due and due-at-delay scheduling coverage - Verifies nack_with_id fixed-delay fallback with backoff configured - Guards batch normalization against duplicate redelivery scheduling Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Asserts DLQ routing stays in payload processing - Confirms negative ack handler has no DLQ identifiers - Verifies nack_with_id dispatches without redelivery count Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Adds named config-parity tests for single, partitioned, multi-topic, regex, and reader paths - Reuses ConsumerConfig construction to avoid broker-dependent tests - Confirms backoff policy and fixed delay survive cloning Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Sets explicit 1ms nack redelivery delay on primary DLQ consumers - Prevents existing broker DLQ tests from inheriting the 60s default - Leaves DLQ consumer builders and production behavior unchanged Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Remove the explicit short negative-ack redelivery delay from broker DLQ tests so they inherit the new 60s parity default. Derive the DLQ receive timeout from the shared default delay constant plus the existing receive cushion. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Make oversized negative-ack redelivery delay validation report the specific failed constraint and include the rejected Duration. Add assertions for the improved error message. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Restore MessageIdDataReceiver to the upstream two-tuple shape - Remove the public MessageIdDataResult alias - Keep optional redelivery count plumbing in pub(crate) internal aliases Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Restore single-topic consumer selection to the upstream count check - Remove reader regex rejection and exactly-one-topic error behavior - Keep reader construction on direct TopicConsumer::new plumbing Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Inline TopicConsumer::new construction in builder and regex refresh paths - Remove the factory and config clone production seams - Delete helper-seam tests rather than preserving source-shape checks Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Remove the extra Consumer::nack_redelivery_delay accessor - Drop the Debug bound from NegativeAckBackoff implementors - Add manual Debug output for internal config and tracker state Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Remove internal engine close message and closed flag - Restore TopicConsumer close to direct broker close - Revert unacked ticker stop/drop lifecycle redesign while keeping nack cleanup Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Restore direct ack sender flow and exact unacked cleanup - Keep negative-ack tracker cancellation for acked entries - Narrow engine test seam to negative-ack redelivery dispatch only Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Configure DLQ consumers with an explicit 1ms nack redelivery delay - Remove the 60s default-derived broker receive timeout helper - Keep the default delay proof at the tracker unit-test layer Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Delete include_str-based implementation shape tests from consumer modules - Keep behavior and type-level tests for nack message and engine behavior - Leave PR-scope evidence in the upstream-check artifact Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
abbudao
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds negative-acknowledgment redelivery delay and backoff support, implementing PIP-106.
Before this change, Rust
nackeffectively requested immediate broker redelivery. Other Pulsar clients already support delayed negative-ack redelivery, and Go/Java default to a 60s fixed delay. This PR brings the Rust client closer to those established semantics while keeping the public API idiomatic for this crate.This contribution was tested in production at my company for two weeks and so far we haven't noticed any issues.
⚠ Breaking behavior change
Default negative-ack redelivery delay changed from 0 ms to 60 seconds (Go/Java parity).
Consumers that relied on immediate redelivery on negative ack will now wait 60 seconds
before the message is redelivered.
Suggested release note:
Semver implication: this changes observable consumer behavior for any caller using
Consumer::nackorConsumer::nack_with_idwithout configuring a custom delay — treat asa minor breaking change for crates.io semver purposes.
Design references
The design was checked against the existing client implementations:
Consumer.Nack/NackID,negativeAcksTracker, andConsumerOptions.NackRedeliveryDelay/NackBackoffPolicy.ConsumerImpl.negativeAcknowledge,NegativeAcksTracker, andnegativeAckRedeliveryDelay(...)/negativeAckRedeliveryBackoff(...).The Rust API mirrors the same concepts with builder methods and
Durationvalues rather than copying either client API directly.User-facing changes
ConsumerBuilder::with_nack_redelivery_delay(Duration).NegativeAckBackoff/MultiplierRedeliveryBackoffsupport for message-based negative acknowledgments.redelivery_countintoMessage::redelivery_count()and internal negative-ack scheduling.README.mdandexamples/negative_ack.rs.Implementation notes for reviewers
Backoff is only applied to message-based nacks:
That path has access to the broker redelivery count, so it can compute the configured backoff delay. ID-only nacks do not carry that count:
For ID-only nacks, the client falls back to the configured fixed delay, or the 60s default. This follows the same practical limitation called out in the Go client.
The internal channel types were split deliberately. The broker redelivery count enters through
CommandMessage, then needs to travel fromConsumerEnginetoTopicConsumersoMessage::redelivery_count()and message-based nack backoff can work. Changing the existing public-facingMessageIdDataReceiveralias from(MessageIdData, Payload)to(MessageIdData, Payload, Option<u32>)would have been a source-breaking tuple-shape change, so the wider tuple is kept behindpub(crate)asInternalMessageIdDataResult.The same applies to engine messages.
EngineMessage::Nack(MessageIdData)is still the public shape; the newInternalEngineMessage::Nack(MessageIdData, Option<u32>)carries redelivery count only inside the consumer implementation.InternalEngineEventalso keeps the private negative-ack timer event out of the public enum.The delayed redelivery state is tracked client-side and dispatched by asking the broker to redeliver due message ids. The tracker normalizes batch-specific ids to the broker redelivery entry while still tracking individual pending batch indexes so later acks can cancel the right scheduled nacks.
A normal ack or cumulative ack cancels matching pending negative-ack redelivery state. This avoids redelivering a message that was nacked and then successfully processed before the delayed redelivery request fired.
Out of scope
This PR deliberately does not change ack-timeout backoff, DLQ/retry-letter semantics, regex topic matching, reader behavior, or broad consumer lifecycle behavior.